package com.ruoyi.datastatistics.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.ruoyi.basedata.domain.Concentrator;
import com.ruoyi.basedata.domain.RedisKeyName;
import com.ruoyi.basedata.domain.TerminalsXb;
import com.ruoyi.basedata.mapper.ConcentratorMapper;
import com.ruoyi.basedata.mapper.TerminalsXbMapper;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.common.utils.JsonUtils;
import com.ruoyi.datastatistics.Agreement.QianBaoTong;
import com.ruoyi.datastatistics.Agreement.QianBaoTongProactive;
import com.ruoyi.datastatistics.domain.MeterHistoryData;
import com.ruoyi.datastatistics.domain.RealtimegatherdataXb;
import com.ruoyi.datastatistics.domain.TerminalHistoryData;
import com.ruoyi.datastatistics.model.QianBaoTongMessageModel;
import com.ruoyi.datastatistics.model.QianBaoTongProactiveModel;
import com.ruoyi.datastatistics.service.IMeterHistoryDataService;
import com.ruoyi.datastatistics.service.IRealtimegatherdataXbService;
import com.ruoyi.datastatistics.service.QianBaoTongService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * 千宝通协议服务实现层
 *
 * @author TsingYu
 * @version 1.0
 * @date 2024/11/9 8:56
 */
@Service
@Slf4j
public class QianBaoTongServiceImpl implements QianBaoTongService {

    private final IRealtimegatherdataXbService realtimegatherdataXbService;
    private final IMeterHistoryDataService meterHistoryDataService;
    private final ConcentratorMapper concentratorMapper;
    private final RedisCache redisCache;
    private final TerminalsXbMapper terminalsXbMapper;

    public QianBaoTongServiceImpl(IRealtimegatherdataXbService realtimegatherdataXbService, IMeterHistoryDataService meterHistoryDataService, ConcentratorMapper concentratorMapper, RedisCache redisCache, TerminalsXbMapper terminalsXbMapper) {
        this.realtimegatherdataXbService = realtimegatherdataXbService;
        this.meterHistoryDataService = meterHistoryDataService;
        this.concentratorMapper = concentratorMapper;
        this.redisCache = redisCache;
        this.terminalsXbMapper = terminalsXbMapper;
    }

    private static final String FACTORY_CODE = "1020";

    /**
     * MQ消息订阅回调
     *
     * @param message 消息
     */
    @Override
    public void callbackSubscribe(String message) {

        try {
            QianBaoTongMessageModel messageModel = JsonUtils.jsonToObject(message, QianBaoTongMessageModel.class);
            log.info("千宝通订阅消息：{}", JsonUtils.objectToJson(messageModel));
            if (CollectionUtils.isNotEmpty(messageModel.getUpDataList())) {
                int i = 1;
                for (QianBaoTongMessageModel.UpData up : messageModel.getUpDataList()) {
                    log.info("第{}次循环", i);
                    i++;
                    byte[] data = QianBaoTong.objectToByteArray(up.getUpData());
                    QianBaoTongProactiveModel<Object> commonPart = QianBaoTongProactive.decodeCommonPart(data);
                    if (commonPart.getAfn() == QianBaoTongProactiveModel.Afn.REQUESTING && (new QianBaoTongProactiveModel.Fn(0, 57)).equals(commonPart.getFn())) {
                        QianBaoTongProactiveModel<QianBaoTongProactiveModel.Copy> model = new QianBaoTongProactiveModel<>(commonPart, QianBaoTongProactive.decodeCopyData(commonPart.getDataUnitRaw()));
                        renewXbHistoryAndRealTimeData(model, messageModel.getPushTime());
                    }

                }
            }

        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }


    }

    /**
     * 更新户表实时和历史数据
     *
     * @param model model
     */
    private void renewXbHistoryAndRealTimeData(QianBaoTongProactiveModel<QianBaoTongProactiveModel.Copy> model, long pushTime) {
        QianBaoTongProactiveModel.Copy copyData = model.getDataUnit();
        Instant now = Instant.ofEpochMilli(pushTime);
        // 有表数据的话插入到历史数据表和实时数据表中
        if (copyData.getMeterData() != null && copyData.getMeterData().length > 0) {
            List<MeterHistoryData> historyDataList = new ArrayList<>(copyData.getMeterData().length);
            List<RealtimegatherdataXb> realTimeDataList = new ArrayList<>(copyData.getMeterData().length);
            // 查找集中器id
            LambdaQueryWrapper<Concentrator> wrapper = Wrappers.lambdaQuery();
            wrapper.eq(Concentrator::getConcentratorId, model.getAddress().getAddress())
                    .eq(Concentrator::getDelFlag, 0)
                    .eq(Concentrator::getSupplierId, FACTORY_CODE)
                    .last("limit 1");
            Long concentratorId = concentratorMapper.selectOne(wrapper).getId();
            for (QianBaoTongProactiveModel.MeterData meter : copyData.getMeterData()) {
                String meterAddr = concentratorId + "-" + meter.getSeq();
                TerminalsXb terminalsXb = redisCache.getCacheMapValue(RedisKeyName.terminalInfo + FACTORY_CODE, meterAddr);
                terminalsXb = terminalsXb == null ? terminalsXbMapper.selectOne(new LambdaQueryWrapper<TerminalsXb>().eq(TerminalsXb::getCommunicateAddress, meterAddr).eq(TerminalsXb::getTerminalMark, 1).eq(TerminalsXb::getDelFlag, 0).last("limit 1")) : terminalsXb;
                if (terminalsXb != null) {
                    if (checkIsSaveInRedis(meterAddr, pushTime)) {
                        continue;
                    }
                    MeterHistoryData historyData = new MeterHistoryData();
                    historyData.setTenantId(terminalsXb.getTenantId());
                    historyData.setLocationNb(terminalsXb.getLocationNb());
                    historyData.setSupplierId(terminalsXb.getSupplierId());
                    historyData.setTerminalId(terminalsXb.getTerminalId());
                    historyData.setTotalFlow(meter.getTotalFlow());
                    historyData.setReadDate(LocalDateTime.ofInstant(now, ZoneId.systemDefault()).format(DateUtils.YYYY_MM_DD_HH_MM_SS_FORMATTER));
                    historyData.setReadNumber(meter.getTotalFlow());
                    historyData.setReverseNumber(BigDecimal.ZERO);
                    historyData.setValveState(meter.getValveState() == 1 ? 0 : 1);
                    historyData.setCreateDate(new Date());
                    historyDataList.add(historyData);
                    redisCache.SortedSetAdd(RedisKeyName.backupDB_HistoricalData + FACTORY_CODE + "_" + meterAddr, historyData, pushTime);
                    RealtimegatherdataXb realtimegatherdata = new RealtimegatherdataXb();
                    realtimegatherdata.setLocationNb(terminalsXb.getLocationNb());
                    List<RealtimegatherdataXb> isUpdate = realtimegatherdataXbService.selectRealtimegatherdataXbList(realtimegatherdata);
                    RealtimegatherdataXb realtimegatherdataXb = CollectionUtils.isNotEmpty(isUpdate) ? isUpdate.get(0) : null;
                    if (realtimegatherdataXb == null) {
                        realtimegatherdataXb = new RealtimegatherdataXb();
                        realtimegatherdataXb.setCreateTime(new Date());
                    }
                    realtimegatherdataXb.setTerminalId(terminalsXb.getTerminalId());
                    realtimegatherdataXb.setLocationNb(terminalsXb.getLocationNb());
                    realtimegatherdataXb.setTenantId(terminalsXb.getTenantId());
                    realtimegatherdataXb.setValveState(new BigDecimal(historyData.getValveState()));
                    realtimegatherdataXb.setReadDate(Date.from(now));
                    realtimegatherdataXb.setTotalFlow(meter.getTotalFlow());
                    realTimeDataList.add(realtimegatherdataXb);
                } else {
                    log.error("Redis中未找到表地址{}的户表信息", meterAddr);
                }
            }
            if (CollectionUtils.isNotEmpty(historyDataList)) {
                meterHistoryDataService.batchinsertMeterHistoryData(historyDataList);
            }
            if (CollectionUtils.isNotEmpty(realTimeDataList)) {
                realtimegatherdataXbService.batchinsertRealtimegatherdataXb(realTimeDataList);
            }
        }


    }

    /**
     * 校验该条历史数据是否重复
     */
    public boolean checkIsSaveInRedis(String communicateAddress, long time) {
        Set<TerminalHistoryData> list = redisCache.SortedSetRangeByRank(RedisKeyName.backupDB_HistoricalData + FACTORY_CODE + "_" + communicateAddress, time, time);
        return CollectionUtils.isNotEmpty(list);
    }
}
